package vip.liux.contracts.infrastructure.localmsg;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.Setter;
import org.apache.commons.lang3.ClassUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.reflect.MethodUtils;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import vip.liux.contracts.shared.kits.ApplicationKit;
import vip.liux.core.utils.JsonUtils;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * 本地消息切面，拦截被 @LocalMessage 注解的方法，实现消息记录、异步/同步执行、重试等功能
 */
@Aspect
public class LocalMessageAspect {

    /**
     * Jackson对象映射器，用于参数序列化和反序列化
     */
    public static final ObjectMapper OBJECT_MAPPER = JsonUtils.getObjectMapper();

    /**
     * 重试间隔（分钟）
     */
    public static final int RETRY_INTERVAL_MINUTES = 2;
    private static final Logger log = LoggerFactory.getLogger(LocalMessageAspect.class);
    /**
     * 线程池执行器，用于异步执行本地消息
     */
    @Autowired
    private ThreadPoolTaskExecutor taskExecutor;
    /**
     * 本地消息仓库，默认使用内存实现，可通过Setter替换
     */
    @Setter
    @Autowired(required = false)
    private LocalMessageRepository repository = LocalMessageRepository.INSTANCE;

    @Scheduled(initialDelay = 5L * 1000, fixedRate = 2L * 60 * 1000)
    public void compensation() {
        log.debug("[local message]compensation start");
        loadWaitRetryRecords().forEach(this::doAsyncInvoke);
    }

    /**
     * 切面方法，拦截被 @LocalMessage 注解的方法
     *
     * @param joinPoint    切点
     * @param localMessage 注解实例
     * @return 返回值（本实现返回null）
     * @throws Throwable 异常
     */
    @Around("@annotation(localMessage)")
    public Object doAspect(ProceedingJoinPoint joinPoint, LocalMessage localMessage) throws Throwable {

        Method method = ((MethodSignature) joinPoint.getSignature()).getMethod();

        if (InvokeStatusHolder.inInvoke()) {
            return joinPoint.proceed();
        }

        boolean async = localMessage.async();

        // 获取方法参数类型列表
        List<String> params = Arrays.stream(method.getParameterTypes())
                .map(Class::getName).toList();

        // 构建调用上下文快照
        InvokeCtx ctx = InvokeCtx.builder()
                .className(method.getDeclaringClass().getName())
                .methodName(method.getName())
                .paramTypes(JsonUtils.toJson(params))
                .args(JsonUtils.toJson(joinPoint.getArgs()))
                .build();

        // 构建本地消息记录
        LocalMessageRecord record = new LocalMessageRecord(
                JsonUtils.toJson(ctx),
                localMessage.maxRetryTimes(),
                offsetTimestamp((long) RETRY_INTERVAL_MINUTES)
        );

        log.info("record local message, record: {}, async {}", JsonUtils.toJson(record), async);
        invoke(record, async);
        return null;
    }

    /**
     * 保存消息记录，并根据事务状态决定何时执行
     *
     * @param record 消息记录
     * @param async  是否异步
     */
    private void invoke(LocalMessageRecord record, boolean async) {
        save(record);
        boolean inTx = TransactionSynchronizationManager.isActualTransactionActive();
        if (inTx) {
            // 如果在事务中，提交后再执行
            TransactionSynchronizationManager.registerSynchronization(
                    new TransactionSynchronization() {
                        @Override
                        public void afterCompletion(int status) {
                            if (status == TransactionSynchronization.STATUS_COMMITTED) execute(record, async);
                        }
                    }
            );
        } else {
            // 非事务直接执行
            execute(record, async);
        }
    }

    /**
     * 保存消息记录
     */
    private void save(LocalMessageRecord record) {
        repository.save(record);
    }

    private List<LocalMessageRecord> loadWaitRetryRecords() {
        // 查询 RETRY_INTERVAL_MINUTES 分钟之前需要重试的任务
        // 加入 create_time 查询条件，避免刚入库的数据被查询出来
        return repository.loadWaitRetryRecords(
                List.of(LocalMessageRecord.TaskStatus.RETRY.name(), LocalMessageRecord.TaskStatus.INIT.name()),
                System.currentTimeMillis(),
                RETRY_INTERVAL_MINUTES
        );
    }

    /**
     * 执行本地消息
     *
     * @param record 消息记录
     * @param async  是否异步
     */
    private void execute(LocalMessageRecord record, boolean async) {
        if (async) {
            doAsyncInvoke(record);
        } else {
            doInvoke(record);
        }
    }

    /**
     * 异步执行消息
     */
    private void doAsyncInvoke(LocalMessageRecord record) {
        taskExecutor.execute(() -> doInvoke(record));
    }

    /**
     * 实际执行消息调用
     * 反射还原方法并调用，处理成功、失败、重试等逻辑
     */
    private void doInvoke(LocalMessageRecord record) {
        String snapshot = record.getReqSnapshot();
        if (StringUtils.isBlank(snapshot)) {
            log.warn("Request snapshot is blank, recordId: {}", record.getId());
            invokeFail(record, "Request snapshot is blank");
            return;
        }

        InvokeCtx ctx = JsonUtils.toObject(snapshot, InvokeCtx.class);
        try {
            InvokeStatusHolder.startInvoke();

            // 反射获取目标Bean和方法
            Class<?> target = ClassUtils.getClass(ctx.getClassName());
            Object bean = ApplicationKit.getBean(target);

            List<Class<?>> paramTypes = getParamTypes(OBJECT_MAPPER.readValue(ctx.getParamTypes(), new TypeReference<>() {
            }));
            Method method = MethodUtils.getMatchingMethod(target, ctx.getMethodName(), paramTypes.toArray(new Class[0]));
            Object[] args = getArgs(paramTypes, ctx.getArgs());

            // 调用目标方法
            method.invoke(bean, args);

            invokeSuccess(record);
        } catch (ClassNotFoundException e) {
            log.error("Class not found for invocation, className:{}, recordId:{}", ctx.getClassName(), record.getId(), e);
            invokeFail(record, e.getMessage());
        } catch (IllegalArgumentException e) {
            log.error("Argument illegal for invocation, className:{}, recordId:{}", ctx.getClassName(), record.getId(), e);
            invokeFail(record, e.getMessage());
        } catch (InvocationTargetException e) {
            // 这里拿到原始异常
            Throwable cause = e.getCause();
            String errorMsg = cause != null ? cause.getMessage() : "Unknown error";
            log.error("Invocation failed: {}.{}, recordId: {}, cause: {}",
                    ctx.getClassName(), ctx.getMethodName(), record.getId(), errorMsg, cause);
            retry(record, errorMsg);
        } catch (Throwable e) {
            log.error("invocation failed, className:{}, recordId:{}", ctx.getClassName(), record.getId(), e);
            retry(record, e.getMessage());
        } finally {
            InvokeStatusHolder.endInvoke();
        }
    }

    /**
     * 调用成功，更新状态为SUCCESS
     */
    private void invokeSuccess(LocalMessageRecord record) {
        record.setStatus(LocalMessageRecord.TaskStatus.SUCCESS.name());
        repository.updateById(record);
    }

    /**
     * 调用失败，更新状态为FAIL并记录原因
     */
    private void invokeFail(LocalMessageRecord record, String errorMsg) {
        record.setStatus(LocalMessageRecord.TaskStatus.FAIL.name());
        record.setFailReason(errorMsg);
        repository.updateById(record);
    }

    /**
     * 调用异常，判断是否需要重试，更新状态和重试次数
     *
     * @param record   消息记录
     * @param errorMsg 错误信息
     */
    private void retry(LocalMessageRecord record, String errorMsg) {
        Integer retryTimes = record.getRetryTimes() + 1;
        LocalMessageRecord updateDO = new LocalMessageRecord();
        updateDO.setId(record.getId());
        updateDO.setFailReason(errorMsg);
        updateDO.setNextRetryTime(getNextRetryTime(retryTimes));
        if (retryTimes >= record.getMaxRetryTimes()) {
            updateDO.setStatus(LocalMessageRecord.TaskStatus.FAIL.name());
        } else {
            updateDO.setRetryTimes(retryTimes);
            updateDO.setStatus(LocalMessageRecord.TaskStatus.RETRY.name());
        }
        repository.updateById(updateDO);
    }

    /**
     * 反序列化参数类型字符串为Class对象列表
     *
     * @param params 参数类型名列表
     * @return Class对象列表
     */
    private List<Class<?>> getParamTypes(List<String> params) {
        return params.stream()
                .map(name -> {
                    try {
                        return Class.forName(name);
                    } catch (ClassNotFoundException e) {
                        log.warn("Parameter class not found: {}", name, e);
                        throw new IllegalArgumentException("Parameter class not found: " + name, e);
                    }
                })
                .collect(Collectors.toList());
    }

    /**
     * 反序列化参数值
     *
     * @param paramTypes 参数类型
     * @param argsJson   参数值的JSON
     * @return 参数对象数组
     * @throws IOException 反序列化异常
     */
    private Object[] getArgs(List<Class<?>> paramTypes, String argsJson) throws IOException {
        List<Object> args = OBJECT_MAPPER.readValue(argsJson, new TypeReference<>() {
        });
        return IntStream.range(0, paramTypes.size())
                .mapToObj(i -> convertArg(paramTypes.get(i), args.get(i)))
                .toArray();
    }

    /**
     * 参数类型转换，确保参数类型与目标方法一致
     *
     * @param targetType 目标类型
     * @param arg        参数值
     * @return 转换后的参数
     */
    private Object convertArg(Class<?> targetType, Object arg) {
        if (targetType.isAssignableFrom(arg.getClass())) {
            return arg;
        }
        try {
            byte[] paramBytes = OBJECT_MAPPER.writeValueAsBytes(arg);
            return OBJECT_MAPPER.readValue(paramBytes, targetType);
        } catch (IOException e) {
            log.error("Failed to convert argument: {}", arg, e);
            throw new IllegalArgumentException("Argument conversion failed", e);
        }
    }

    /**
     * 根据重试次数计算下一次重试的时间
     * 本方法通过重试次数计算出需要等待的分钟数，从而确定下一次重试的具体时间
     * 这是为了避免短时间内多次重试，通过增加等待时间来逐步减少重试频率
     *
     * @param retryTimes 重试次数，用于计算等待时间
     * @return 下一次重试的时间戳（毫秒）
     */
    private Long getNextRetryTime(Integer retryTimes) {
        // 计算等待分钟数，使用重试间隔的指数增长来增加等待时间
        double waitMinutes = Math.pow(RETRY_INTERVAL_MINUTES, retryTimes);
        return offsetTimestamp((long) waitMinutes);
    }

    /**
     * 获取当前时间偏移指定分钟后的时间戳
     *
     * @param minutes 偏移分钟数
     * @return 偏移后的时间戳（毫秒）
     */
    private Long offsetTimestamp(Long minutes) {
        return System.currentTimeMillis() + minutes * 60 * 1000;
    }
}